//package dataStream;
//
//import org.apache.flink.api.common.functions.RuntimeContext;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
//import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
//import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
//import org.apache.http.HttpHost;
//import org.elasticsearch.action.index.IndexRequest;
//import org.elasticsearch.client.Requests;
//
//import java.util.ArrayList;
//import java.util.HashMap;
//import java.util.List;
//import java.util.Map;
//
///**
// * @desc
// * @Author xuyouchang
// * @Date 2021/8/5
// * @Update 2021/8/5 14:40
// */
//public class EsSinkBasic {
//    public static void main(String[] args) throws Exception {
//        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        List<HttpHost> httpHosts = new ArrayList<>();
//        httpHosts.add(new HttpHost("172.17.1.101", 9200, "http"));
//        httpHosts.add(new HttpHost("172.17.1.102", 9200, "http"));
//
//        DataStream<String> input = env.readTextFile("E:\\test_data\\kafkaSink.txt");
//
//        ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
//                httpHosts,
//                new ElasticsearchSinkFunction<String>() {
//                    public IndexRequest createIndexRequest(String element) {
//                        Map<String, String> json = new HashMap<>();
//                        json.put("data", element);
//
//                        return Requests.indexRequest()
//                                .index("flink-sink-index")
//                                .source(json);
//                    }
//
//                    @Override
//                    public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
//                        indexer.add(createIndexRequest(element));
//                    }
//                }
//        );
//
//        input.addSink(esSinkBuilder.build());
//        env.execute();
//    }
//}
